package com.cml.rpc.register;

import com.cml.rpc.loadbalance.LoadBalance;
import com.cml.rpc.loadbalance.RoundLoadBalance;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;

import java.net.InetSocketAddress;
import java.util.List;

/**
 * zookeeper实现服务注册
 * @author cml
 * @create 2021/7/22 10:04
 */
public class ZkServiceRegister implements ServiceRegister{
    // curator提供的zookeeper客户端
    private CuratorFramework client;
    // 负载方式
    private LoadBalance loadBalance;
    // zookeeper根路径节点
    private static final String ROOT_PATH = "myRpc";

    // 负责zookeeper的初始化
    public ZkServiceRegister() {
        // 默认负载方式
        this(new RoundLoadBalance());
    }

    public ZkServiceRegister(LoadBalance loadBalance) {

        // 指数时间重试
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        // 不管是服务提供者，消费者都需要与zookeeper建立联系
        this.client = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181")
                .sessionTimeoutMs(40000).retryPolicy(retryPolicy).namespace(ROOT_PATH).build();
        this.client.start();
        this.loadBalance = loadBalance;
        System.out.println("zookeeper链接成功");
    }
    @Override
    public void register(String serviceName, InetSocketAddress serverAddress) {
        try {
            // 创建永久节点，服务提供者下线时，不删除服务名，只删除地址
            if (client.checkExists().forPath("/" + serviceName) == null) {
                client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT)
                        .forPath("/" + serviceName);
            }
            String path = "/" + serviceName + "/" + getServiceAddress(serverAddress);
            // 创建临时节点
            client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL)
                    .forPath(path);
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("此服务已存在");
        }
    }

    @Override
    public InetSocketAddress serviceDiscover(String serviceName) {
        try {
            List<String> strings = client.getChildren().forPath("/" + serviceName);
//            // 默认使用第一个，后面扩展负载均衡
//            String address = strings.get(0);
            String address = loadBalance.loadBalance(strings);
            return parseAddress(address);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    // 字符串解析为地址
    private InetSocketAddress parseAddress(String address) {
        String[] result = address.split(":");
        return new InetSocketAddress(result[0], Integer.parseInt(result[1]));
    }

    // 地址： xxx.xxx.xxx.xxx:port
    private String getServiceAddress(InetSocketAddress serverAddress) {
        return serverAddress.getHostName() + ":" + serverAddress.getPort();
    }
}
